package cn.chrilwe.pigpig.im.service;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;

import cn.chrilwe.pigpig.core.base.Code;
import cn.chrilwe.pigpig.im.config.ServerConfig;
import cn.chrilwe.pigpig.im.model.RabbitModel;
import cn.chrilwe.pigpig.im.protocol.TalkRequest;
import cn.chrilwe.pigpig.im.protocol.TalkResponse;
import cn.chrilwe.pigpig.im.rabbit.common.RabbitEnum;
import cn.chrilwe.pigpig.im.rabbit.factory.RabbitSession;
import cn.chrilwe.pigpig.im.rabbit.factory.RabbitSessionFactory;
import cn.chrilwe.pigpig.im.service.executor.TaskExecutor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

/**
 * 
 * @author chrilwe
 * 2020-8-12
 */
@Service
public class MessageDispatcherService {
	
	@Autowired
	private StringRedisTemplate stringRedisTemplate;
	
	@Autowired
	private MessageService messageService;
	
	@Autowired
	private ServerConfig serverConfig;
	
	@Autowired
	private RabbitSessionFactory rabbitSessionFactory;
	

	//分发消息到指定client
	public void dispactcherMessage(Map<String, Map> groupChannel, TalkRequest talkRequest, ChannelHandlerContext ctx) {
		topicMessage(groupChannel, talkRequest, ctx);
	}
	
	protected void topicMessage(Map<String, Map> groupChannel,TalkRequest talkRequest, ChannelHandlerContext ctx) {
		TalkResponse res = messageService.topicMsg(talkRequest);
		List<String> shards = serverConfig.getShards();
		if(shards != null && shards.size() > 0) {
			TaskExecutor.getInstance().executor(new Runnable() {

				@Override
				public void run() {
					sendMqMsg(talkRequest);
				}
				
			});
		}
		if(res.getStatus() == Code.SUCCESS) {
			Map<String, Channel> map = groupChannel.get(talkRequest.getRoomId()+"");
			Set<Entry<String, Channel>> entrySet = map.entrySet();
			for (Entry<String, Channel> entry : entrySet) {
				String clientId = entry.getKey();
				Channel ch = entry.getValue();
				ch.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(res)));
			}
		} else {
			ctx.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(res)));
		}
	}
	
	protected void sendMqMsg(TalkRequest request) {
		String serverName = serverConfig.getServerName();
		List<String> shards = serverConfig.getShards();
		if(shards != null && shards.size() > 0) {
			for(String shard: shards) {
				realSend(JSON.toJSONString(request), shard, shard);
			}
		}
	}
	
	protected void realSend(String message, String queue, String routingKey) {
		RabbitSession session = rabbitSessionFactory.openSession();
		RabbitModel rabbitModel = new RabbitModel();
		rabbitModel.setAutoAck(true);
		rabbitModel.setAutoDelete(false);
		rabbitModel.setCurrentConnectPrivate(false);
		rabbitModel.setExchange(RabbitEnum.EXCHANGE.getName());
		rabbitModel.setMap(null);
		rabbitModel.setPersistence(true);
		rabbitModel.setQueue(queue);
		rabbitModel.setRoutingKey(routingKey);
		session.send(message, rabbitModel);
	}
}
